package com.mango.ch11;

import com.mango.Tools.DateUtils;
import com.mango.ch04.MyPairComparator;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import scala.Tuple2;

import java.text.SimpleDateFormat;
import java.util.*;

public class Spark_MarkvoJob {
    private static String ROOTPATH = "d:/HadoopData/Spark_MarkvoJob";
    private static String MR1_INPUT = ROOTPATH+"/input";
    private static String MR1_OUTPUT = ROOTPATH+"/step1_output";
    private static String MR2_OUTPUT = ROOTPATH+"/step2_output";
    private static String MR3_OUTPUT = ROOTPATH+"/step3_output";

    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        //提交都本地Master运行spark程序
        /*        conf.setMaster("local");*/
        conf.setAppName("Spark_MarkvoJob");
        conf.setMaster("local");
        conf.set("spark.yarn.dist.files", "src\\yarn-site.xml");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> lines = jsc.textFile(MR1_INPUT, 1).coalesce(9);
        JavaPairRDD<String, Tuple2<Date, Double>> pairs = lines.mapToPair(new PairFunction<String, String, Tuple2<Date, Double>>() {
            @Override
            public Tuple2<String, Tuple2<Date, Double>> call(String s) throws Exception {
                String[] tokens = s.split(",");
                String key = tokens[0];
                SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd");
                Date trancationTime = sf.parse(tokens[2]);
                Double amount = Double.valueOf(tokens[3]);
                Tuple2 t = new Tuple2(trancationTime, amount);
                return new Tuple2<>(key, t);
            }
        });
        pairs.saveAsTextFile(ROOTPATH + "/2");
        //对键值对规约一下
        JavaPairRDD<String, Iterable<Tuple2<Date, Double>>> kValues = pairs.groupByKey();
        kValues.saveAsTextFile(ROOTPATH + "/3");

        //产生状态序列
        JavaPairRDD<String, List<String>> timeSeqence = kValues.mapValues(new Function<Iterable<Tuple2<Date, Double>>, List<String>>() {
            @Override
            public List<String> call(Iterable<Tuple2<Date, Double>> tuple2s) throws Exception {
                List<Tuple2<Date, Double>> list = makeToList(tuple2s);
                Collections.sort(list, new Spark_MarkvoJob.myTupleComparator());
                //转换成状态序列
                List<String> stateSequence = toStateSequence(list);
                return stateSequence;
            }
        });
        timeSeqence.saveAsTextFile(ROOTPATH + "/4");
        //生成马尔可夫状态转移矩阵
        //key 为[（s1,s2),1] 从状态1转移到状态2
        JavaPairRDD<Tuple2<String, String>, Integer> model = timeSeqence.flatMapToPair(new PairFlatMapFunction<Tuple2<String, List<String>>, Tuple2<String, String>, Integer>() {
            @Override
            public Iterator<Tuple2<Tuple2<String, String>, Integer>> call(Tuple2<String, List<String>> stringListTuple2) throws Exception {
                List<String> states = stringListTuple2._2;
                //状态序列少于两个，不能形成状态转化  所以不成立
                if (states.size() < 2) return new ArrayList<Tuple2<Tuple2<String, String>, Integer>>().iterator();
                List<Tuple2<Tuple2<String, String>, Integer>> tmpList = new ArrayList<>();
                for (int i = 0; i < states.size() - 1; i++) {
                    String fromState = states.get(i);
                    String toState = states.get(i + 1);
                    tmpList.add(new Tuple2<Tuple2<String, String>, Integer>(new Tuple2<>(fromState, toState), 1));
                }
                return tmpList.iterator();
            }
        });
        model.saveAsTextFile(ROOTPATH + "/5");
        //组合规约(fromState,toState)的频度 (自定义规约 规则，实现自己的规约逻辑)
        JavaPairRDD<Tuple2<String, String>, Integer> tmpresult = model.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });
        tmpresult.saveAsTextFile(ROOTPATH + "/6");
        //输出模型需要的格式
        JavaRDD<String> modelFormat = tmpresult.map(new Function<Tuple2<Tuple2<String, String>, Integer>, String>() {
            @Override
            public String call(Tuple2<Tuple2<String, String>, Integer> tuple2IntegerTuple2) throws Exception {
                StringBuilder sb = new StringBuilder();
                sb.append(tuple2IntegerTuple2._1._1);
                sb.append(",");
                sb.append(tuple2IntegerTuple2._1._2);
                sb.append("\t");
                sb.append(tuple2IntegerTuple2._2);
                return sb.toString();
            }
        });
        modelFormat.saveAsTextFile(ROOTPATH + "/7");
        jsc.close();
    }

    static List<String> toStateSequence(List<Tuple2<Date, Double>> list) {
        if (list == null || list.size() < 2) return new ArrayList<>();
        List<String> tmpList = new ArrayList<>();
        for (int i = 0; i < list.size() - 1; i++) {
            Date endDate = list.get(i)._1;
            Date nowDate = list.get(i + 1)._1;
            Long dayDiff = DateUtils.getDatePoor(endDate, nowDate);
            double preAmount = list.get(i)._2;
            double nextAmount = list.get(i + 1)._2;
            String dd = null;
            String ad = null;
            if (dayDiff < 30)
                dd = "S";
            else if (dayDiff < 60)
                dd = "M";
            else
                dd = "L";
            if (preAmount < 0.9 * nextAmount)
                ad = "L";
            else if (preAmount < 1.1 * nextAmount)
                ad = "E";
            else
                ad = "G";
            tmpList.add(dd + ad);
        }
        return tmpList;
    }

    static List<Tuple2<Date, Double>> makeToList(Iterable<Tuple2<Date, Double>> tuple2s) {
        List<Tuple2<Date, Double>> tmpList = new ArrayList<>();
        for (Tuple2<Date, Double> t :
                tuple2s) {
            tmpList.add(t);
        }
        return tmpList;
    }


    static class myTupleComparator implements Comparator<Tuple2<Date, Double>> {

        @Override
        public int compare(Tuple2<Date, Double> o1, Tuple2<Date, Double> o2) {
            int result = (int) (o1._1.getTime() - o2._1.getTime());
            return result;
        }
    }
}
